Concurrent handling via FuturesUnordered breaks subscribe→unsubscribe ordering assumptions and can leak subscriptions

The new FuturesUnordered-based loop makes handle_msg futures for all message types run concurrently. Because of how handle_msg is structured:

• AccountSubscribe:

  • Calls Self::add_sub(...).await.
  • add_sub inserts the AccountSubscription into subs before its first .await (account_subscribe().await).

• AccountUnsubscribe:

  • Has no .await; its entire branch is synchronous and completes on the first poll of its future.

With FuturesUnordered (and without any explicit ordering guarantees), there is now a valid interleaving where:

1. A subscribe message for pubkey is received first and its handle_msg future is pushed into pending_messages.
2. A subsequent unsubscribe message for the same pubkey is received and its handle_msg future is also pushed.
3. When pending_messages.next() runs, it may poll the unsubscribe future before the subscribe future’s first poll.
4. In that case:
   • The unsubscribe branch checks subs.lock().get(&pubkey) and finds no entry yet, so it returns AccountSubscriptionDoesNotExist.
   • Afterwards, the subscribe future is polled for the first time, and add_sub inserts the AccountSubscription into the map and sets up the real subscription.

This results in:

• The caller logically issuing subscribe then unsubscribe, but:
  • Seeing an unsubscribe error (AccountSubscriptionDoesNotExist).
  • Still ending up with a live subscription (because add_sub completes later and spawns the update task).
• A leaked subscription that the caller believes was removed.
• The comment in add_sub about “messages to this actor are processed in the order they are sent” no longer being true under the concurrent FuturesUnordered processing.

This is a semantic correctness regression introduced by the concurrency change, not just a theoretical concern.

Consider one of the following fixes:

• Restrict concurrency to subscription establishment only:

  • Keep the actor loop sequential for control messages (AccountUnsubscribe, Reconnect).
  • For AccountSubscribe, enqueue/track only the long‑running “establish subscription” work in FuturesUnordered (e.g., a dedicated async function that performs account_subscribe + spawn of the update loop), while the main handle_msg path that updates subs and returns the oneshot response preserves message ordering.

• Explicitly track in‑flight subscriptions:

  • Maintain a per‑Pubkey state for “subscription being created”.
  • Treat an unsubscribe for an in‑flight subscription as success and arrange for the creation path to observe and honor that cancellation (e.g., cancel before or immediately after adding to subs).

Additionally, on shutdown or when messages_receiver is closed, the loop breaks and drops pending_messages, abruptly canceling in‑flight handle_msg futures. This is a behavior change from the previous sequential implementation, where the currently executing handle_msg would finish before exit. Please double‑check that it’s acceptable for callers to sometimes see pending subscribe/reconnect requests fail with dropped oneshots during shutdown.

Also applies to: 232-252, 257-257, 362-370, 376-381

🌐 Web query:

Does `futures_util::stream::FuturesUnordered` provide any guarantee about the order in which newly pushed futures are first polled, relative to each other? Specifically, can a future that was pushed later be polled and complete before a previously pushed future’s first poll?

💡 Result:

Short answer: No — there is no guarantee. FuturesUnordered is unordered: pushing a future does not poll it, and futures are driven only when they generate wake notifications, so a future pushed later can be polled (and even complete) before an earlier-pushed future’s first poll. Use FuturesOrdered if you need FIFO semantics. [1][2]

Sources:

• futures-util/FuturesUnordered docs (behaviour and note that it “may complete in any order”). [1]
• futures::stream::FuturesOrdered docs (contrast: preserves FIFO order). [2]